package com.hyl.msg;

import java.util.Properties;

import org.apache.kafka.clients.producer.Callback;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

public class KfkProducer{

	private KafkaProducer<byte[],byte[]> _producer;
	private String _namesrvAddr = "localhost:9020";
	private String _groupid = "testp";

//String kclass,String vclass,
	public KfkProducer(String group, String serverurl) {
		
		_namesrvAddr = serverurl;
		_groupid = group;
		Properties props = KfkConfig.getPubConf();
		if (serverurl != null)
			props.put("bootstrap.servers", _namesrvAddr);
		if (group != null)
			props.put("group.id", _groupid);

//		if(kclass.equals("String") ) {
//			props.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
//		}else if(kclass.equals("byte[]")) {
//			props.put("key.serializer","org.apache.kafka.common.serialization.ByteArraySerializer");
//		}
//		if(vclass.equals("String") ) {
//			props.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
//		}else if(vclass.equals("byte[]")) {
//			props.put("value.serializer","org.apache.kafka.common.serialization.ByteArraySerializer");
//		}
		_producer = new KafkaProducer<byte[],byte[]>(props);
	}

	public void pushMsg(String topic, byte[] key, byte[] data) {
		_producer.send(new ProducerRecord<byte[],byte[]>(topic, key, data));

	}

	public void close() {
		_producer.flush(); // 将缓冲区的消息立即发送
		// consumer.commit(); //消费者手动确认消费进度
		_producer.close();
	}

	public void pushMsgAsyn(String topic, byte[] key, byte[] data) {
		ProducerRecord<byte[],byte[]> pr = new ProducerRecord<byte[],byte[]>(topic, key, data);
		_producer.send(pr, new Callback() {
			// 回调函数
			@Override
			public void onCompletion(RecordMetadata metadata, Exception exception) {
				if (null != exception) {
					System.out.println("记录的offset在:" + metadata.offset());
					System.out.println(exception.getMessage() + exception);
				}
			}

		});
	}

}
